package com.atbeijing.bigdata.spark.mytest.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


/**
 * Map[品类，HotCategory]
 *
 */
object Top10_2 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top")
    val sc = new SparkContext(conf)

    //先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。
    val file: RDD[String] = sc.textFile("data/user_visit_action.txt")
    // 创建累加器
    val acc = new HotCategoryAcc
    // 注册累加器
    sc.register(acc, "HotCategory1")

    // 累加器的执行代码一般用在行动算子中，而不是转换算子
    file.foreach(
      line => {
        val datas = line.split("_")
        if ( datas(6) != "-1" ) {
          // 点击
          acc.add((datas(6), "click"))
        } else if ( datas(8) != "null" ) {
          // 下单
          val ids = datas(8).split(",")
          ids.foreach(
            id => {
              acc.add((id, "order"))
            }
          )
        } else if (datas(10) != "null") {
          // 支付
          val ids = datas(10).split(",")
          ids.foreach(
            id => {
              acc.add((id, "pay"))
            }
          )
      }
    })

    // 获取累加器的值 ,将他映射成集合,随后转为list
    val counts: List[HotCategory] = acc.value.map(_._2).toList

    val top10: List[HotCategory] = counts.sortWith(
      (left, right) => {
        if (left.clickCount > right.clickCount) {
          true
        } else if (left.clickCount == right.clickCount) {
          if (left.orderCount > right.orderCount) {
            true
          } else if (left.orderCount == right.orderCount) {
            left.payCount > right.payCount
          } else {
            false
          }
        } else {
          false
        }
      }
    ).take(10)
    //top10.foreach(println)

    println("===================================================")
    // TODO 需求二 : Top10热门品类中每个品类的Top10活跃Session统计
    file.cache()
    //过滤=>保留品类ID在top10 中的数据
    val filtered: RDD[String] = file.filter(line => {
      val data: Array[String] = line.split("_")
      val ids: List[String] = top10.map(hc => {
        hc.categoryId
      })
      ids.contains(data(6))
    })
    //整理数据结构((品类,session),1)
    val r1: RDD[((String, String), Int)] = filtered.map(line => {
      val data: Array[String] = line.split("_")
      ((data(6), data(2)), 1)
    })
    //聚合((品类,session),sum)
    val r2: RDD[((String, String), Int)] = r1.reduceByKey(_+_)
    //整理数据结构(品类,(session,sum))
    val r3: RDD[(String, (String, Int))] = r2.map(w => {
      (w._1._1, (w._1._2, w._2))
    })
    //按照品类聚合,(品类,iterable[(session,sum)])
    val r4: RDD[(String, Iterable[(String, Int)])] = r3.groupByKey()
    //降序,取session前十
    r4.map(w=>{
      val tp10: List[(String, Int)] = w._2.toList.sortBy(s =>s._2)(Ordering.Int.reverse).take(10)
      (w._1,tp10)
    }).collect().foreach(println)

  }

  //样例类存数据
  case class HotCategory(categoryId:String, var clickCount:Long,var  orderCount:Long,var payCount:Long)

  //自定义累加器,设置IN(品类id,行为),OUT(key为品类id,value为HotCategory的map)
  class HotCategoryAcc extends AccumulatorV2 [(String,String),mutable.Map[String,HotCategory]] {
    //创建容器
    private val map: mutable.Map[String, HotCategory] = mutable.Map[String,HotCategory]()

    override def isZero: Boolean = map.isEmpty

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {new HotCategoryAcc()}

    override def reset(): Unit = map.clear()

    //v(品类id,行为)
    override def add(v: (String, String)): Unit = {
      //根据品类id查询,若已存在,更新hotCategory对象的属性值,若不存在,给个初始值
      val hotCategory: HotCategory = map.getOrElse(v._1,HotCategory(v._1,0L,0L,0L))
      v._2 match {
        case "click" => hotCategory.clickCount += 1
        case "order" => hotCategory.orderCount += 1
        case "pay" =>hotCategory.payCount += 1
      }
      map.update(v._1,hotCategory)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      //other一个HotCategoryAcc对象,调用它的value方法
      val mapOther: mutable.Map[String, HotCategory] = other.value
      //将other合并到this中
      mapOther.foreach {
        case (categoryId,hotCategory) => {
          //获取当前对象的HotCategory
          val thisCategory: HotCategory =map.getOrElse(categoryId,HotCategory(categoryId,0,0,0))
          //将other的属性添加进来
          thisCategory.clickCount += hotCategory.clickCount
          thisCategory.orderCount += hotCategory.orderCount
          thisCategory.payCount += hotCategory.payCount
          map.update(categoryId,thisCategory)
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = map
  }




}
